package com.chiaching.framwork.utils;

import com.chiaching.framwork.config.RocketMqConfig;
import com.chiaching.framwork.core.exception.BusException;
import com.chiaching.framwork.core.handler.mq.DefaultMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.stream.Collectors;

/**
 * @author jiajing.he
 * @date 2021/5/20 下午2:46
 */
@ConditionalOnBean(RocketMqConfig.class)
@Component
@Slf4j
public class RocketUtils {


    @Resource
    private DefaultMQProducer defaultMQProducer;
    @Resource
    private DefaultMQPushConsumer defaultMQPushConsumer;

    public void subscribe(String topic, String tags, DefaultMessageListener messageListener){
        RocketMqConfig.MESSAGE_LISTENERS.put(topic,messageListener);
        try {
            defaultMQPushConsumer.subscribe(topic, tags);
        } catch (MQClientException e) {
            RocketMqConfig.MESSAGE_LISTENERS.remove(topic);
            log.info("subscribe topic:[{}],tags:[{}] error ......",topic,tags);
        }
        log.info("topic:[{}]，tags[{}] 监听注册成功......",topic,tags);
    }

    /**
     * @param topic
     * @param tags 多个tag用 , 进行分隔
     * @param body
     * @return
     */
    public SendResult send(String topic, String tags, String body) {
        try {
            SendResult send = defaultMQProducer.send(Arrays.stream(tags.split(",")).map(e -> new Message(topic, e, body.getBytes())).collect(Collectors.toList()));
            log.info("rocketmq send success,param[topic:{},tags:{},body:{}] result:[{}]", topic, tags, body, send);
            return send;
        } catch (Exception e) {
            log.error("rocketmq send error, param[topic:{},tags:{},body:{}]", topic, tags, body, e);
        }
        throw new BusException("rocketmq send error");
    }





    public void sendCallBack(String topic, String tags, String body, SendCallback sendCallback) {
        try {
            defaultMQProducer.send(new Message(topic, tags, body.getBytes()),sendCallback);
            log.info("rocketmq sendCallBack success,param[topic:{},tags:{},body:{}]", topic, tags, body);
        } catch (Exception e) {
            log.error("rocketmq send error, param[topic:{},tags:{},body:{}]", topic, tags, body, e);
            throw new BusException("rocketmq sendCallBack error");
        }
    }


    /**
     * @param topic
     * @param tags
     * @param body
     * @param delayLevel "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h" 对应rocket每个延迟等级的时间
     * @return
     */
    public SendResult sendDelay(String topic, String tags, String body, int delayLevel) {
        try {
            Message message = new Message(topic, tags, body.getBytes());
            message.setDelayTimeLevel(delayLevel);
            SendResult send = defaultMQProducer.send(message);
            log.info("rocketmq sendDelay success,param[topic:{},tags:{},body:{},delayLevel:{}] result:[{}]", topic, tags, body,delayLevel, send);
            return send;
        } catch (Exception e) {
            log.error("rocketmq sendDelay error, param[topic:{},tags:{},body:{}]", topic, tags, body, e);
        }
        throw new BusException("rocketmq sendDelay error");
    }

}
